package com.codejiwei.flink.cep;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/7/31
 * desc:
 **/
public class Flink_CEP_Test {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        DataStreamSource<String> sourceDS = env.readTextFile("E:\\workspaces_sdses\\bigdata-flink\\src\\main\\resources\\data\\ticker.txt");


        Schema schema = Schema.newBuilder()
                .column("symbol", DataTypes.STRING())
                .column("rowtime", DataTypes.TIMESTAMP())
                .column("price", DataTypes.DOUBLE())
                .column("tax", DataTypes.INT())
                .build();
//
//
//        tEnv.createTemporaryTable("sourceTable", TableDescriptor.forConnector("filesystem")
//                .schema(schema)
//                .option("path", "src/main/resources/data/ticker.txt")
//                .format(FormatDescriptor.forFormat("json")
////                        .option("filed.delimiter", ",")
//                        .build())
//                .build());


        tEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "  symbol STRING,\n" +
                "  rowtime TIMESTAMP(3) ,\n" +
                "  price DOUBLE,\n" +
                "  tax INT, \n" +
                "  WATERMARK FOR rowtime AS rowtime - INTERVAL '5' SECOND \n" +
                ") WITH (\n" +
                "  'connector' = 'filesystem',\n" +
                "  'path' = 'src/main/resources/data/ticker.txt',\n" +
                "  'format' = 'json'\n" +
                ")");


        Schema schema1 = Schema.newBuilder()
                .column("symbol", DataTypes.STRING())
                .column("start_tstamp", DataTypes.TIMESTAMP())
                .column("end_tstamp", DataTypes.TIMESTAMP())
                .column("avgPrice", DataTypes.DOUBLE())
                .build();


        tEnv.createTemporaryTable("sinkTable", TableDescriptor.forConnector("print")
                .schema(schema1)
                .build());


        tEnv.executeSql("insert into sinkTable SELECT *\n" +
                "FROM sourceTable\n" +
                "    MATCH_RECOGNIZE (\n" +
                "        PARTITION BY symbol\n" +
                "        ORDER BY rowtime\n" +
                "        MEASURES\n" +
                "            FIRST(A.rowtime) AS start_tstamp,\n" +
                "            LAST(A.rowtime) AS end_tstamp,\n" +
                "            AVG(A.price) AS avgPrice\n" +
                "        ONE ROW PER MATCH\n" +
                "        AFTER MATCH SKIP PAST LAST ROW\n" +
                "        PATTERN (A+ B)\n" +
                "        DEFINE\n" +
                "            A AS AVG(A.price) < 15\n" +
                "    ) MR");


    }
}
